package basicknowledge;

import java.util.ArrayDeque;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static basicknowledge.CallableAndFuture.SerialExecutor.CORE_POOL_SIZE;

/**
 * Created by Zonda on 2017/1/12.
 * Java 中的 Callable 和 Future 示例
 */
public class CallableAndFuture {

    public static void main(String[] args) {

        testFutureTask();

        testExecutorService();

        testCompletionService();
    }

    private static void testCompletionService() {

        //ExecutorService threadPool = Executors.newSingleThreadExecutor();
        //Executor threadPool = THREAD_POOL_EXECUTOR;
        Executor threadPool = new SerialExecutor();
        System.out.println("CORE_POOL_SIZE: " + CORE_POOL_SIZE);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(threadPool);
        for (int i = 1; i < 5; i++) {
            final int taskId = i;
            completionService.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    System.out.println("testCompletionService taskId : " + taskId + " call is run before!");
                    Thread.sleep(5000);
                    System.out.println("testCompletionService taskId : " + taskId + " call is run after!");
                    return taskId;
                }
            });

//            completionService.submit(() -> {
//                System.out.println("testCompletionService taskId : " + taskId + " call is run before!");
//                Thread.sleep(5000);
//                System.out.println("testCompletionService taskId : " + taskId + " call is run after!");
//                return taskId;
//            });

        }

        System.out.println("testCompletionService is run 1----");

        for (int i = 1; i < 5; i++) {
            try {
                System.out.println("testCompletionService take: " + completionService.take().get() + " flag: " + i);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        System.out.println("testCompletionService is run 2----");
    }

    static class SerialExecutor implements Executor {

        static final Executor THREAD_POOL_EXECUTOR;

        private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
        // We want at least 2 threads and at most 4 threads in the core pool,
        // preferring to have 1 less than the CPU count to avoid saturating
        // the CPU with background work
        static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
        private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
        private static final int KEEP_ALIVE_SECONDS = 30;

        private static final ThreadFactory sThreadFactory = new ThreadFactory() {
            private final AtomicInteger mCount = new AtomicInteger(1);

            public Thread newThread(Runnable r) {
                return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
            }
        };

        private static final BlockingQueue<Runnable> sPoolWorkQueue =
                new LinkedBlockingQueue<>(128);

        static {
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
                    sPoolWorkQueue, sThreadFactory);
            threadPoolExecutor.allowCoreThreadTimeOut(true);
            THREAD_POOL_EXECUTOR = threadPoolExecutor;
        }

        final ArrayDeque<Runnable> mTasks = new ArrayDeque<>();
        Runnable mActive;

        SerialExecutor() {

        }

        //这里就是说明如何保证任务按队列次序执行的
        public synchronized void execute(final Runnable r) {

            mTasks.offer(new Runnable() {
                @Override
                public void run() {
                    try {
                        r.run();
                    } finally {
                        scheduleNext();
                    }
                }
            });

//            mTasks.offer(() -> {
//                try {
//                    r.run();
//                } finally {
//                    scheduleNext();
//                }
//            });

            if (mActive == null) {
                scheduleNext();
            }
        }

        synchronized void scheduleNext() {
            if ((mActive = mTasks.poll()) != null) {
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }
    }


    private static void testExecutorService() {

        ExecutorService threadPool = Executors.newCachedThreadPool();
        Future<Integer> future = threadPool.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                System.out.println("testExecutorService callable call is run! ");
                return new Random().nextInt(100);
            }
        });

//        Future<Integer> future = threadPool.submit(() -> {
//            System.out.println("testExecutorService callable call is run! ");
//            return new Random().nextInt(100);
//        });

        try {
            Thread.sleep(2000);
            System.out.println("testExecutorService: " + future.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    private static void testFutureTask() {

        Callable<Integer> callable = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {

                System.out.println("testFutureTask callable call is run!");
                return new Random().nextInt(100);
            }
        };
//        Callable<Integer> callable = () -> {
//            System.out.println("testFutureTask callable call is run!");
//            return new Random().nextInt(100);
//        };

        FutureTask<Integer> futureTask = new FutureTask<>(callable);

        new Thread(futureTask).start();

        try {
            Thread.sleep(2000);
            System.out.println("testFutureTask: " + futureTask.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}
